package com.omni.monitor.plugin.threadpool;

import com.alibaba.ttl.threadpool.TtlExecutors;
import com.omni.monitor.model.threadpool.CustomRejectedExecutionHandler;
import com.omni.monitor.model.threadpool.ThreadPoolStats;
import com.omni.monitor.model.threadpool.ThreadPoolTrend;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.lang.reflect.Field;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;
@Slf4j
@Component
public class ThreadPoolMonitor {
    // 被监控的线程池映射表
    private final Map<String, MonitoredThreadPool> threadPoolMap = new ConcurrentHashMap<>();
    // 历史数据，用于趋势分析
    private final Map<String, List<ThreadPoolStats>> historyStats = new ConcurrentHashMap<>();
    // 任务耗时记录（key：poolName，value：最近100个任务的耗时）
    private final Map<String, List<Long>> taskCostRecords = new ConcurrentHashMap<>();

    @Value("${threadpool.monitor.enabled:true}")
    private boolean monitorEnabled;
    @Value("${threadpool.alert.threshold:80}")
    private int alertThreshold;
    @Value("${threadpool.alert.channels:slack,email}")
    private String alertChannels;
    @Value("${threadpool.history.limit:100}")
    private int historyLimit; // 历史数据保留条数（可配置）

    @Autowired
    private StrategyManager strategyManager; // 注入策略管理器
    // @Autowired
    // private AlertService alertService; // 告警服务（需自行实现）

    @PostConstruct
    public void init() {
        log.info("线程池监控服务启动，预警阈值: {}%, 告警通道: {}, 历史数据保留条数: {}",
                alertThreshold, alertChannels, historyLimit);
    }

    /**
     * 注册线程池到监控系统（修复TTL包装+拒绝处理器）
     */
    public <T extends ThreadPoolExecutor> ThreadPoolExecutor register(String name, T executor) {
        if (!monitorEnabled) {
            return executor;
        }

        MonitoredThreadPool monitoredPool = new MonitoredThreadPool(name,  executor);

        threadPoolMap.put(name, monitoredPool);
        historyStats.put(name, new ArrayList<>());
        taskCostRecords.put(name, new ArrayList<>());

        log.info("线程池 [{}] 已注册到监控系统, 核心线程数: {}, 最大线程数: {}, 队列类型: {}, 队列容量: {}",
                name, executor.getCorePoolSize(), executor.getMaximumPoolSize(),
                executor.getQueue().getClass().getSimpleName(), getQueueCapacity(executor.getQueue()));
        return monitoredPool;
    }

    /**
     * 定时收集线程池统计数据（增强：策略检查+派生字段计算）
     */
    @Scheduled(fixedRate = 10000) // 每10秒执行一次
    public void collectStats() {
        if (!monitorEnabled || threadPoolMap.isEmpty()) {
            return;
        }

        threadPoolMap.forEach((name, pool) -> {
            ThreadPoolStats stats = new ThreadPoolStats();
            ThreadPoolExecutor executor = pool.getOriginalExecutor();
            BlockingQueue<?> queue = executor.getQueue();

            // 1. 基础指标采集
            stats.setTimestamp(System.currentTimeMillis());
            stats.setPoolName(name);
            stats.setActiveThreads(executor.getActiveCount());
            stats.setCorePoolSize(executor.getCorePoolSize());
            stats.setMaximumPoolSize(executor.getMaximumPoolSize());
            stats.setLargestPoolSize(executor.getLargestPoolSize());
            stats.setPoolSize(executor.getPoolSize());
            stats.setQueueSize(queue.size());
            stats.setQueueCapacity(getQueueCapacity(queue));
            stats.setTaskCount(executor.getTaskCount());
            stats.setCompletedTaskCount(executor.getCompletedTaskCount());
            stats.setRejectedCount(pool.getRejectedCount());

            // 2. 计算派生字段（空闲线程、剩余队列容量等）
            stats.calculateDerivedFields(); // 计算idleThreads等派生字段
            stats.calculateMetrics();       // 计算activeThreadRatio等比例指标
            // 3. 计算核心指标（比例类）
            calculateMetrics(stats);
            // 4. 计算平均任务执行时间（基于历史任务耗时）
            calculateAvgTaskExecuteTime(name, stats);

            // 5. 存储历史数据（控制条数）
            List<ThreadPoolStats> history = historyStats.get(name);
            history.add(stats);
            if (history.size() > historyLimit) {
                history.remove(0);
            }

            // 6. 检查告警
            checkAlert(stats);
            // 7. 检查并执行策略
            strategyManager.checkAndExecuteStrategy(name, stats, executor);

            // 日志输出
            log.debug("线程池 [{}] 状态: 活跃线程 {}/{} ({}%), 空闲线程 {}, 队列 {}/{} ({}%), 已完成任务 {}, 拒绝任务 {}, 平均执行时间 {:.1f}ms",
                    name, stats.getActiveThreads(), stats.getMaximumPoolSize(), stats.getActiveThreadRatio(),
                    stats.getIdleThreads(), stats.getQueueSize(), stats.getQueueCapacity(), stats.getQueueUsageRatio(),
                    stats.getCompletedTaskCount(), stats.getRejectedCount(), stats.getAvgTaskExecuteTime());
        });
    }

    /**
     * 计算平均任务执行时间
     */
    private void calculateAvgTaskExecuteTime(String poolName, ThreadPoolStats stats) {
        List<Long> costList = taskCostRecords.get(poolName);
        if (costList.isEmpty()) {
            stats.setAvgTaskExecuteTime(0);
            return;
        }
        double avg = costList.stream().mapToLong(Long::longValue).average().orElse(0);
        stats.setAvgTaskExecuteTime(avg);
        // 保留最近100个任务耗时
        if (costList.size() > 100) {
            costList.subList(0, costList.size() - 100).clear();
        }
    }

    /**
     * 计算核心指标（比例类）
     */
    private void calculateMetrics(ThreadPoolStats stats) {
        // 活跃线程比例
        stats.setActiveThreadRatio(stats.getMaximumPoolSize() > 0
                ? (double) stats.getActiveThreads() / stats.getMaximumPoolSize() * 100
                : 0);
        // 队列使用率
        stats.setQueueUsageRatio(stats.getQueueCapacity() > 0
                ? (double) stats.getQueueSize() / stats.getQueueCapacity() * 100
                : 0);
        // 任务完成率
        stats.setTaskCompletionRatio(stats.getTaskCount() > 0
                ? (double) stats.getCompletedTaskCount() / stats.getTaskCount() * 100
                : 100);
    }

    /**
     * 检查告警（需自行实现AlertService）
     */
    private void checkAlert(ThreadPoolStats stats) {
        boolean needAlert = false;
        StringBuilder alertMsg = new StringBuilder();

        if (stats.getActiveThreadRatio() > alertThreshold) {
            needAlert = true;
            alertMsg.append("活跃线程比例过高: ").append(String.format("%.1f%%", stats.getActiveThreadRatio())).append("; ");
        }
        if (stats.getQueueUsageRatio() > alertThreshold) {
            needAlert = true;
            alertMsg.append("队列使用率过高: ").append(String.format("%.1f%%", stats.getQueueUsageRatio())).append("; ");
        }
        if (stats.getRejectedCount() > 0 && stats.getRejectedCount() > historyStats.get(stats.getPoolName()).get(0).getRejectedCount()) {
            needAlert = true;
            alertMsg.append("新增拒绝任务: ").append(stats.getRejectedCount()).append("个; ");
        }

        if (needAlert) {
            String finalMsg = String.format("线程池告警 [%s]: %s", stats.getPoolName(), alertMsg);
            // alertService.sendAlert(finalMsg, alertChannels.split(",")); // 发送告警
            log.warn(finalMsg);
        }
    }

    /**
     * ---------------------- 手动配置方法（供Controller调用） ----------------------
     */
    // 设置核心线程数
    public void setCorePoolSize(String poolName, int targetCoreSize) {
        MonitoredThreadPool pool = threadPoolMap.get(poolName);
        if (pool == null) {
            throw new IllegalArgumentException("线程池不存在：" + poolName);
        }
        ThreadPoolExecutor executor = pool.getOriginalExecutor();
        if (targetCoreSize < 0 || targetCoreSize > executor.getMaximumPoolSize()) {
            throw new IllegalArgumentException("核心线程数需满足 0 ≤ 目标值 ≤ 最大线程数");
        }
        long oldCoreSize = executor.getCorePoolSize();
        executor.setCorePoolSize(targetCoreSize);
        log.info("手动调整线程池[{}]核心线程数：{} -> {}", poolName, oldCoreSize, targetCoreSize);
    }

    // 设置最大线程数
    public void setMaximumPoolSize(String poolName, int targetMaxSize) {
        MonitoredThreadPool pool = threadPoolMap.get(poolName);
        if (pool == null) {
            throw new IllegalArgumentException("线程池不存在：" + poolName);
        }
        ThreadPoolExecutor executor = pool.getOriginalExecutor();
        int cpuCore = Runtime.getRuntime().availableProcessors();
        targetMaxSize = Math.min(targetMaxSize, cpuCore * 10); // 限制最大为CPU×10
        if (targetMaxSize < executor.getCorePoolSize()) {
            throw new IllegalArgumentException("最大线程数需 ≥ 核心线程数");
        }
        long oldMaxSize = executor.getMaximumPoolSize();
        executor.setMaximumPoolSize(targetMaxSize);
        log.info("手动调整线程池[{}]最大线程数：{} -> {}", poolName, oldMaxSize, targetMaxSize);
    }

    // 设置队列容量（仅支持有界队列）
    public void setQueueCapacity(String poolName, int targetQueueCap) {
        MonitoredThreadPool pool = threadPoolMap.get(poolName);
        if (pool == null) {
            throw new IllegalArgumentException("线程池不存在：" + poolName);
        }
        ThreadPoolExecutor executor = pool.getOriginalExecutor();
        BlockingQueue<?> queue = executor.getQueue();
        if (!(queue instanceof java.util.concurrent.ArrayBlockingQueue)) {
            log.warn("仅支持ArrayBlockingQueue队列容量调整");
        }
        if (targetQueueCap <= 0) {
            throw new IllegalArgumentException("队列容量需 > 0");
        }
        // 队列容量无法直接修改，需重建队列（此处为示例，实际需结合业务停止任务后重建）
        log.warn("线程池[{}]队列容量调整：{} -> {}（需手动重建队列）", poolName, queue.size(), targetQueueCap);
    }

    /**
     * ---------------------- 工具方法 ----------------------
     */
    // 获取指定线程池最新统计
    public ThreadPoolStats getThreadPoolStats(String threadPoolName) {
        List<ThreadPoolStats> history = historyStats.get(threadPoolName);
        return history != null && !history.isEmpty() ? history.get(history.size() - 1) : null;
    }

    // 获取所有线程池最新统计
    public List<ThreadPoolStats> getAllPoolStats() {
        return threadPoolMap.keySet().stream()
                .map(this::getThreadPoolStats)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    }

    // 获取线程池趋势
    public ThreadPoolTrend getTrend(String poolName) {
        List<ThreadPoolStats> history = historyStats.getOrDefault(poolName, Collections.emptyList());
        if (history.isEmpty()) {
            return null;
        }

        ThreadPoolTrend trend = new ThreadPoolTrend();
        trend.setPoolName(poolName);
        trend.setDataCount(history.size());
        // 平均活跃线程比例
        trend.setAvgActiveThreadRatio(history.stream()
                .mapToDouble(ThreadPoolStats::getActiveThreadRatio)
                .average()
                .orElse(0));
        // 平均队列使用率
        trend.setAvgQueueUsageRatio(history.stream()
                .mapToDouble(ThreadPoolStats::getQueueUsageRatio)
                .average()
                .orElse(0));
        // 总拒绝任务数
        trend.setTotalRejectedCount(history.stream()
                .mapToLong(ThreadPoolStats::getRejectedCount)
                .max()
                .orElse(0));
        // 平均任务完成率
        trend.setTaskCompletionRatio(history.stream()
                .mapToDouble(ThreadPoolStats::getTaskCompletionRatio)
                .average()
                .orElse(100));
        return trend;
    }

    // 获取队列容量
    private int getQueueCapacity(BlockingQueue<?> queue) {
        try {
            if (queue instanceof java.util.concurrent.LinkedBlockingQueue) {
                Field field = java.util.concurrent.LinkedBlockingQueue.class.getDeclaredField("capacity");
                field.setAccessible(true);
                return (int) field.get(queue);
            } else if (queue instanceof java.util.concurrent.ArrayBlockingQueue) {
                Field field = java.util.concurrent.ArrayBlockingQueue.class.getDeclaredField("items");
                field.setAccessible(true);
                Object[] items = (Object[]) field.get(queue);
                return items.length;
            }
        } catch (Exception e) {
            log.warn("获取队列容量失败", e);
        }
        return Integer.MAX_VALUE; // 无界队列默认最大值
    }

    /**
     * 监控包装的线程池（修复拒绝处理器+任务耗时统计）
     */
    public class MonitoredThreadPool extends ThreadPoolExecutor {
        private final String name;
        private final ThreadPoolExecutor originalExecutor;
        private long rejectedCount = 0;

        public MonitoredThreadPool(String name, ThreadPoolExecutor executor) {
            super(
                    executor.getCorePoolSize(),
                    executor.getMaximumPoolSize(),
                    executor.getKeepAliveTime(java.util.concurrent.TimeUnit.NANOSECONDS),
                    java.util.concurrent.TimeUnit.NANOSECONDS,
                    executor.getQueue(),
                    executor.getThreadFactory(),
                    new CustomRejectedExecutionHandler() // 替换为监控拒绝处理器
            );
            this.name = name;
            this.originalExecutor = executor;
        }

        /**
         * 重写execute方法，统计任务执行时间
         */
        @Override
        public void execute(Runnable command) {
            long startTime = System.currentTimeMillis();
            try {
                super.execute(() -> {
                    try {
                        command.run();
                    } finally {
                        // 记录任务耗时
                        long cost = System.currentTimeMillis() - startTime;
                        taskCostRecords.get(name).add(cost);
                    }
                });
            } catch (Exception e) {
                log.error("线程池[{}]执行任务异常", name, e);
                throw e;
            }
        }

        public ThreadPoolExecutor getOriginalExecutor() {
            return originalExecutor;
        }

        public long getRejectedCount() {
            return rejectedCount;
        }
    }
}
